/*-------------------------------------------------------------------------
 *
 * walreceiver.h
 *	  从 replication/walreceiverfuncs.c 导出的内容。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * src/include/replication/walreceiver.h
 *
 *-------------------------------------------------------------------------
 */
#ifndef _WALRECEIVER_H
#define _WALRECEIVER_H

#include "access/xlog.h"
#include "access/xlogdefs.h"
#include "getaddrinfo.h"		/* 用于 NI_MAXHOST */
#include "pgtime.h"
#include "port/atomics.h"
#include "replication/logicalproto.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
#include "storage/latch.h"
#include "storage/spin.h"
#include "utils/tuplestore.h"

/* 用户可设置的参数 */
extern PGDLLIMPORT int wal_receiver_status_interval;
extern PGDLLIMPORT int wal_receiver_timeout;
extern PGDLLIMPORT bool hot_standby_feedback;

/*
 * MAXCONNINFO：连接字符串的最大大小。
 *
 * XXX：这是否应该移动到 pg_config_manual.h 中？
 */
#define MAXCONNINFO		1024

/* 我们是否可以允许备用接收来自另一个备用的复制连接？ */
#define AllowCascadeReplication() (EnableHotStandby && max_wal_senders > 0)

/*
 * WalRcv->walRcvState 的值。
 */
typedef enum
{
	WALRCV_STOPPED,				/* 停止并且不能再次启动 */
	WALRCV_STARTING,			/* 已启动，但进程尚未
								 * 初始化 */
	WALRCV_STREAMING,			/* walreceiver 正在流式传输 */
	WALRCV_WAITING,				/* 停止流式传输，等待命令 */
	WALRCV_RESTARTING,			/* 请求重新启动流式传输 */
	WALRCV_STOPPING				/* 请求停止，但仍在运行 */
} WalRcvState;

/* 管理 walreceiver 进程的共享内存区 */
typedef struct
{
	/*
	 * 当前活动 walreceiver 进程的 PID、其当前状态和
	 * 启动时间（实际上是请求启动的时间）。
	 */
	pid_t		pid;
	WalRcvState walRcvState;
	ConditionVariable walRcvStoppedCV;
	pg_time_t	startTime;

	/*
	 * receiveStart 和 receiveStartTLI 表示将要接收的
	 * 第一个字节位置和时间线。当启动进程启动
	 * walreceiver 时，它将这些设置为希望流式传输的起始点。
	 */
	XLogRecPtr	receiveStart;
	TimeLineID	receiveStartTLI;

	/*
	 * flushedUpto-1 是已经接收到的最后一个字节位置，
	 * receivedTLI 是它来自的时间线。在 walreceiver 的第一次启动中，
	 * 这些设置为 receiveStart 和 receiveStartTLI。之后，
	 * walreceiver 会在每次将接收到的 WAL 刷新到
	 * 磁盘时更新这些。
	 */
	XLogRecPtr	flushedUpto;
	TimeLineID	receivedTLI;

	/*
	 * latestChunkStart 是当前接收到的"WAL批次"的起始字节位置。
	 * 实际上它与上一次刷新的 flushedUpto 值相同。
	 * 启动过程可以使用此信息来检测它是否跟上了。
	 */
	XLogRecPtr	latestChunkStart;

	/*
	 * 发送和接收任何接收到的消息的时间。
	 */
	TimestampTz lastMsgSendTime;
	TimestampTz lastMsgReceiptTime;

	/*
	 * 发送方 WAL 的最新报告结束
	 */
	XLogRecPtr	latestWalEnd;
	TimestampTz latestWalEndTime;

	/*
	 * 连接字符串；初始设置为连接到主服务器，后来
	 * 被覆盖以隐藏安全敏感字段。
	 */
	char		conninfo[MAXCONNINFO];

	/*
	 * 活动复制连接的主机名（可以是主机名、IP 地址或目录路径）
	 * 和端口号。
	 */
	char		sender_host[NI_MAXHOST];
	int			sender_port;

	/*
	 * 复制槽名称；还用于 walreceiver 与主服务器的连接
	 */
	char		slotname[NAMEDATALEN];

	/*
	 * 如果这是一个临时复制槽，连接时需要重新创建。
	 */
	bool		is_temp_slot;

	/* 一旦 conninfo 准备好显示（模糊的密码等），则设置为真 */
	bool		ready_to_display;

	/*
	 * 启动进程使用的延迟，用于在告知 walreceiver
	 * 从哪里开始流式传输（在设置 receiveStart 和
	 * receiveStartTLI 之后）后，将其唤醒，并告诉它
	 * 每当应用特别标记的提交记录时向主服务器发送应用反馈。
	 * 这通常在 walreceiver 正在运行时映射到 procLatch。
	 */
	Latch	   *latch;

	slock_t		mutex;			/* 锁定上述共享变量 */

	/*
	 * 类似 flushedUpto，但在写入后和刷新之前进展，
	 * 不需要获取自旋锁。其他进程可以读取到这一点的数据，
	 * 但不应用于数据完整性目的。
	 */
	pg_atomic_uint64 writtenUpto;

	/*
	 * 强制 walreceiver 回复？这不需要锁；排序的内存
	 * 屏障就足够了。但我们确实需要原子获取和
	 * 存储语义，因此使用 sig_atomic_t。
	 */
	sig_atomic_t force_reply;	/* 用作布尔值 */
} WalRcvData;

extern PGDLLIMPORT WalRcvData *WalRcv;

typedef struct
{
	bool		logical;		/* 如果这是逻辑复制流，则为 true，
								 * 如果是物理流，则为 false。 */
	char	   *slotname;		/* 复制槽名称或NULL。 */
	XLogRecPtr	startpoint;		/* 起始点的LSN。 */

	union
	{
		struct
		{
			TimeLineID	startpointTLI;	/* 起始时间线 */
		}			physical;
		struct
		{
			uint32		proto_version;	/* 逻辑协议版本 */
			List	   *publication_names;	/* 发布的字符串列表 */
			bool		binary; /* 请求发布者使用二进制格式 */
			bool		streaming;	/* 大事务的流式传输 */
			bool		twophase;	/* 在准备时流式传输两阶段事务 */
		}			logical;
	}			proto;
#ifdef FDDM
	List *forward_sub_names;
#endif
} WalRcvStreamOptions;

struct WalReceiverConn;
typedef struct WalReceiverConn WalReceiverConn;

/*
 * walreceiver 查询执行的状态。
 *
 * 我们只定义当前使用的状态。
 */
typedef enum
{
	WALRCV_ERROR,				/* 执行查询时发生错误。 */
	WALRCV_OK_COMMAND,			/* 查询执行了实用程序或复制
								 * 命令。 */
	WALRCV_OK_TUPLES,			/* 查询返回了元组。 */
	WALRCV_OK_COPY_IN,			/* 查询开始了COPY FROM。 */
	WALRCV_OK_COPY_OUT,			/* 查询开始了COPY TO。 */
	WALRCV_OK_COPY_BOTH			/* 查询启动了COPY BOTH 复制
								 * 协议。 */
} WalRcvExecStatus;

/*
 * walrcv_exec 的返回值，返回执行状态和
 * 元组（如果有）。
 */
typedef struct WalRcvExecResult
{
	WalRcvExecStatus status;
	int			sqlstate;
	char	   *err;
	Tuplestorestate *tuplestore;
	TupleDesc	tupledesc;
} WalRcvExecResult;

/* WAL 接收器 - libpqwalreceiver 钩子 */

/*
 * walrcv_connect_fn
 *
 * 建立与集群的连接。'logical' 为真时表示
 * 连接为逻辑，假时表示连接为物理。
 * 'appname' 是与连接相关的名称，用于例如
 * fallback_application_name 或 application_name。 返回
 * 关于已建立连接的详细信息，如
 * WalReceiverConn 为每个 WAL 接收模块所定义的。 出错时，
 * 返回 NULL，并包含生成的错误的“err”。
 */
typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
											   bool logical,
											   const char *appname,
											   char **err);

/*
 * walrcv_check_conninfo_fn
 *
 * 解析并验证给定为 'conninfo' 的连接字符串。
 */
typedef void (*walrcv_check_conninfo_fn) (const char *conninfo);

/*
 * walrcv_get_conninfo_fn
 *
 * 返回用户可显示的conninfo字符串。 请注意，任何
 * 安全敏感字段都应该被掩盖。
 */
typedef char *(*walrcv_get_conninfo_fn) (WalReceiverConn *conn);

/*
 * walrcv_get_senderinfo_fn
 *
 * 提供此 WAL 接收器所连接的 WAL 发送器的信息，
 * 如 'sender_host' 为发送者的主机，'sender_port'
 * 为其端口。
 */
typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn,
										  char **sender_host,
										  int *sender_port);

/*
 * walrcv_identify_system_fn
 *
 * 在连接到的集群上运行 IDENTIFY_SYSTEM 并验证
 * 集群的身份。 返回连接的集群的系统ID。
 * 'primary_tli' 是发送者的时间线ID。
 */
typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
											TimeLineID *primary_tli);

/*
 * walrcv_server_version_fn
 *
 * 返回连接的集群的版本号。
 */
typedef int (*walrcv_server_version_fn) (WalReceiverConn *conn);

/*
 * walrcv_readtimelinehistoryfile_fn
 *
 * 从集群获取时间线 'tli' 的时间线历史文件。
 * 返回时间线历史文件的名称为 'filename'，其
 * 内容为 'content' 及其 'size'。
 */
typedef void (*walrcv_readtimelinehistoryfile_fn) (WalReceiverConn *conn,
												   TimeLineID tli,
												   char **filename,
												   char **content,
												   int *size);

/*
 * walrcv_startstreaming_fn
 *
 * 从给定的流式选项启动WAL数据流。 如果连接成功切换到
 * copy-both 模式，则返回true；如果服务器接收命令并成功执行，
 * 但未切换到复制模式，则返回false。
 */
typedef bool (*walrcv_startstreaming_fn) (WalReceiverConn *conn,
										  const WalRcvStreamOptions *options);

/*
 * walrcv_endstreaming_fn
 *
 * 停止WAL数据的流式传输。 返回连接的集群的下一个时间线ID
 * 在 'next_tli' 中，如果没有报告，则返回0。
 */
typedef void (*walrcv_endstreaming_fn) (WalReceiverConn *conn,
										TimeLineID *next_tli);

/*
 * walrcv_receive_fn
 *
 * 接收来自WAL流中的可用消息。 'buffer' 是一个指针
 *，指向持有接收到的消息的缓冲区。 返回数据的长度，
 * 如果尚无数据可用，则返回0（'wait_fd' 是一个可以在重试之前
 * 等待的套接字描述符），如果集群结束了COPY则返回-1。
 */
typedef int (*walrcv_receive_fn) (WalReceiverConn *conn,
								  char **buffer,
								  pgsocket *wait_fd);

/*
 * walrcv_send_fn
 *
 * 向WAL流发送大小为 'nbytes' 的消息，其内容为 'buffer'。
 */
typedef void (*walrcv_send_fn) (WalReceiverConn *conn,
								const char *buffer,
								int nbytes);

/*
 * walrcv_create_slot_fn
 *
 * 创建一个名为 'slotname' 的新复制槽。 'temporary' 定义
 * 槽是否为临时。 'snapshot_action' 定义导出快照所需的行为
 * （请参阅复制协议以获取更多详细信息）。 'lsn' 包含
 * 创建的槽变得一致时的位置的LSN。 对于逻辑
 * 槽，返回导出快照的名称，或对于物理槽返回NULL。
 */
typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
										const char *slotname,
										bool temporary,
										bool two_phase,
#ifdef FDD
										bool failover,
#endif
										CRSSnapshotAction snapshot_action,
										XLogRecPtr *lsn);

/*
 * walrcv_get_backend_pid_fn
 *
 * 返回远程后端进程的 PID。
 */
typedef pid_t (*walrcv_get_backend_pid_fn) (WalReceiverConn *conn);

/*
 * walrcv_exec_fn
 *
 * 发送通用查询（和命令）到远程集群。 'nRetTypes' 是期望返回的属性数量，'retTypes'是包含它们类型OID的数组。 返回执行状态和元组（如果有的话）。
 */
typedef WalRcvExecResult *(*walrcv_exec_fn) (WalReceiverConn *conn,
											 const char *query,
											 const int nRetTypes,
											 const Oid *retTypes);

/*
 * walrcv_disconnect_fn
 *
 * 与集群断开连接。
 */
typedef void (*walrcv_disconnect_fn) (WalReceiverConn *conn);

typedef struct WalReceiverFunctionsType
{
	walrcv_connect_fn walrcv_connect;
	walrcv_check_conninfo_fn walrcv_check_conninfo;
	walrcv_get_conninfo_fn walrcv_get_conninfo;
	walrcv_get_senderinfo_fn walrcv_get_senderinfo;
	walrcv_identify_system_fn walrcv_identify_system;
	walrcv_server_version_fn walrcv_server_version;
	walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
	walrcv_startstreaming_fn walrcv_startstreaming;
	walrcv_endstreaming_fn walrcv_endstreaming;
	walrcv_receive_fn walrcv_receive;
	walrcv_send_fn walrcv_send;
	walrcv_create_slot_fn walrcv_create_slot;
	walrcv_get_backend_pid_fn walrcv_get_backend_pid;
	walrcv_exec_fn walrcv_exec;
	walrcv_disconnect_fn walrcv_disconnect;
} WalReceiverFunctionsType;

extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;

#define walrcv_connect(conninfo, logical, appname, err) \
	WalReceiverFunctions->walrcv_connect(conninfo, logical, appname, err)
#define walrcv_check_conninfo(conninfo) \
	WalReceiverFunctions->walrcv_check_conninfo(conninfo)
#define walrcv_get_conninfo(conn) \
	WalReceiverFunctions->walrcv_get_conninfo(conn)
#define walrcv_get_senderinfo(conn, sender_host, sender_port) \
	WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
#define walrcv_identify_system(conn, primary_tli) \
	WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
#define walrcv_server_version(conn) \
	WalReceiverFunctions->walrcv_server_version(conn)
#define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
	WalReceiverFunctions->walrcv_readtimelinehistoryfile(conn, tli, filename, content, size)
#define walrcv_startstreaming(conn, options) \
	WalReceiverFunctions->walrcv_startstreaming(conn, options)
#define walrcv_endstreaming(conn, next_tli) \
	WalReceiverFunctions->walrcv_endstreaming(conn, next_tli)
#define walrcv_receive(conn, buffer, wait_fd) \
	WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
#define walrcv_send(conn, buffer, nbytes) \
	WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
#ifdef FDD
#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
	WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
#else
#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
	WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
#endif
#define walrcv_get_backend_pid(conn) \
	WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
	WalReceiverFunctions->walrcv_exec(conn, exec, nRetTypes, retTypes)
#define walrcv_disconnect(conn) \
	WalReceiverFunctions->walrcv_disconnect(conn)

static inline void
walrcv_clear_result(WalRcvExecResult *walres)
{
	if (!walres)
		return;

	if (walres->err)
		pfree(walres->err);

	if (walres->tuplestore)
		tuplestore_end(walres->tuplestore);

	if (walres->tupledesc)
		FreeTupleDesc(walres->tupledesc);

	pfree(walres);
}

/* walreceiver.c中的函数原型 */
extern void WalReceiverMain(void) pg_attribute_noreturn();
extern void ProcessWalRcvInterrupts(void);

/* walreceiverfuncs.c中的函数原型 */
extern Size WalRcvShmemSize(void);
extern void WalRcvShmemInit(void);
extern void ShutdownWalRcv(void);
extern bool WalRcvStreaming(void);
extern bool WalRcvRunning(void);
extern void RequestXLogStreaming(TimeLineID tli, XLogRecPtr recptr,
								 const char *conninfo, const char *slotname,
								 bool create_temp_slot);
extern XLogRecPtr GetWalRcvFlushRecPtr(XLogRecPtr *latestChunkStart, TimeLineID *receiveTLI);
extern XLogRecPtr GetWalRcvWriteRecPtr(void);
extern int	GetReplicationApplyDelay(void);
extern int	GetReplicationTransferLatency(void);
extern void WalRcvForceReply(void);

#endif							/* _WALRECEIVER_H */
